package com.ihr360.payroll.service.salarytasklet;

import com.google.common.collect.ImmutableList;
import com.ihr360.common.DateRange;
import com.ihr360.commons.exception.Ihr360Exception;
import com.ihr360.commons.lang.CollectionUtils;
import com.ihr360.expr.ExprArray;
import com.ihr360.job.core.Job;
import com.ihr360.job.core.Step;
import com.ihr360.job.core.config.JobBuilderFactory;
import com.ihr360.job.core.config.StepBuilderFactory;
import com.ihr360.job.core.item.ItemReader;
import com.ihr360.job.core.job.builder.SimpleJobBuilder;
import com.ihr360.job.core.lanuch.support.RunIdIncrementer;
import com.ihr360.job.endpoint.ThreadPoolStatus;
import com.ihr360.payroll.calc.StaffSalaryCalcPeriod;
import com.ihr360.payroll.config.SalaryTaxTableEntity;
import com.ihr360.payroll.dao.ActivePayPeriodRepository;
import com.ihr360.payroll.dao.HistoryPayPeriodRepository;
import com.ihr360.payroll.model.dto.reference.ReferenceService;
import com.ihr360.payroll.model.dto.salary.meta.SalaryField;
import com.ihr360.payroll.model.dto.salary.meta.system.SystemSalaryField;
import com.ihr360.payroll.model.dto.salary.plan.SalaryPlan;
import com.ihr360.payroll.model.dto.salary.plan.SalaryPlanStaff;
import com.ihr360.payroll.model.dto.salary.report.HistoryPayPeriod;
import com.ihr360.payroll.model.dto.salary.report.MergeReportStaffSalary;
import com.ihr360.payroll.model.dto.salary.task.period.ActivePayPeriod;
import com.ihr360.payroll.model.dto.salary.task.period.PayPeriodTerm;
import com.ihr360.payroll.model.dto.salary.task.period.StaffPayPeriodTermCalcContext;
import com.ihr360.payroll.service.salarytasklet.expr.engine.SalaryDependencyEngine;
import com.ihr360.payroll.service.salarytasklet.listener.ItemFailureLoggerListener;
import com.ihr360.payroll.service.salarytasklet.listener.JobCompletionNotificationListener;
import com.ihr360.payroll.service.salarytasklet.model.PreSalaryCalcProcessedResult;
import com.ihr360.payroll.service.salarytasklet.model.SimplePayrollConfig;
import com.ihr360.payroll.service.salarytasklet.step.SalaryCalcContext;
import com.ihr360.payroll.service.salarytasklet.step.processor.AttendancePlanDependencyProcessor;
import com.ihr360.payroll.service.salarytasklet.step.processor.MergeTaxSalaryReportDataProcessor;
import com.ihr360.payroll.service.salarytasklet.step.processor.MergeTaxSalaryReportMetaProcessor;
import com.ihr360.payroll.service.salarytasklet.step.processor.PayrollConfigProcessor;
import com.ihr360.payroll.service.salarytasklet.step.processor.PreSalaryCalcProcessor;
import com.ihr360.payroll.service.salarytasklet.step.processor.SalaryCalcProcessor;
import com.ihr360.payroll.service.salarytasklet.step.processor.SalaryFieldDependencyProcessor;
import com.ihr360.payroll.service.salarytasklet.step.processor.SalaryPlanStaffProcessor;
import com.ihr360.payroll.service.salarytasklet.step.processor.SystemSalaryFieldProcessor;
import com.ihr360.payroll.service.salarytasklet.step.processor.TaxTableProcessor;
import com.ihr360.payroll.service.salarytasklet.step.processor.WorkingCalendarProcessor;
import com.ihr360.payroll.service.salarytasklet.step.reader.CombineSalaryCalcBasedInfoReader;
import com.ihr360.payroll.service.salarytasklet.step.reader.SalaryCalcContextInfoReader;
import com.ihr360.payroll.service.salarytasklet.step.reader.SalaryPayPeriodTermSalaryReader;
import com.ihr360.payroll.service.salarytasklet.step.reader.StepReader;
import com.ihr360.payroll.service.salarytasklet.step.writer.ActivePayPeriodWriter;
import com.ihr360.payroll.service.salarytasklet.step.writer.CalcedStaffSalaryInfoWriter;
import com.ihr360.payroll.service.salarytasklet.step.writer.MergeTaxSalaryReportDataWriter;
import com.ihr360.payroll.service.salarytasklet.step.writer.StaffSalaryCalcContextWriter;
import com.ihr360.sharedservice.model.StatutoryVacation;
import com.ihr360.sharedservice.model.WorkingCalendar;
import com.ihr360.validation.ValidationUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.data.util.Pair;
import org.springframework.stereotype.Component;
import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
import org.springframework.util.Assert;

import java.util.Calendar;
import java.util.List;
import java.util.Map;

/**
 * @author davidwei
 */
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
@Component
public class SalaryCalcJob {
    private static final boolean ALLOW_RESTART_IF_COMPLETED = true;

    private static final int CHUNK_SIZE = 5;
    private static final int NUM_10 = 10;
    @Autowired
    private ActivePayPeriodRepository activePayPeriodRepository;

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Autowired
    private JobCompletionNotificationListener listener;

    @Autowired
    private ItemFailureLoggerListener loggerListener;

    @Autowired
    private SalaryCalcContext salaryCalcContext;

    //Processors

    @Autowired
    private PayrollConfigProcessor payrollConfigProcessor;

    @Autowired
    private AttendancePlanDependencyProcessor attendancePlanDependencyProcessor;

    @Autowired
    private SalaryFieldDependencyProcessor salaryFieldProcessor;

    @Autowired
    private WorkingCalendarProcessor workingCalendarProcessor;

    @Autowired
    private PreSalaryCalcProcessor preSalaryCalcProcessor;

    @Autowired
    private SalaryCalcProcessor salaryCalcProcessor;

    @Autowired
    private MergeTaxSalaryReportMetaProcessor mergeTaxSalaryReportMetaProcessor;

    @Autowired
    private MergeTaxSalaryReportDataProcessor mergeTaxSalaryReportDataProcessor;
    //Reader and writer

    @Autowired
    private StepReader stepReader;

    @Autowired
    private CombineSalaryCalcBasedInfoReader combineSalaryCalcBasedInfoReader;

    @Autowired
    private StaffSalaryCalcContextWriter staffSalaryCalcContextWriter;

    @Autowired
    private SalaryCalcContextInfoReader salaryCalcContextInfoReader;

    @Autowired
    private CalcedStaffSalaryInfoWriter calcedStaffSalaryInfoWriter;

    @Autowired
    private SalaryPayPeriodTermSalaryReader salaryPayPeriodTermSalaryReader;

    @Autowired
    private MergeTaxSalaryReportDataWriter mergeTaxSalaryReportDataWriter;

    @Autowired
    private HistoryPayPeriodRepository historyPayPeriodRepository;

    @Autowired
    private ActivePayPeriodWriter activePayPeriodWriter;

    private ActivePayPeriod payPeriod;

    @Autowired
    private ThreadPoolStatus threadPoolStatus;


    public static final int TOTAL_EXCUTION_COUNT_ALL = 11;
    public static final int TOTAL_EXCUTION_COUNT_PRECALC = 10;
    public static final int TOTAL_EXCUTION_COUNT_CALC = 10;

    private PayPeriodTerm initSalaryContext(SalaryPlan plan, ActivePayPeriod payPeriod, List<String> staffIdList) {
        Assert.notNull(plan, "薪资方案不能为空");
        Assert.notNull(payPeriod, "薪资账期信息不能为空");

        DateRange payrollPeriodRange = DateRange.build(payPeriod.getStartOn(), payPeriod.getEndOn());
        DateRange attendancePeriodRange = DateRange.build(payPeriod.getAttendanceStartOn(), payPeriod.getAttendanceEndOn());


        salaryCalcContext.salaryPlanId = plan.getId();
        salaryCalcContext.payPeriodId = payPeriod.getId();
        salaryCalcContext.attendancePlanId = plan.getAttendancePlan().getId();
        salaryCalcContext.payrollConfigId = plan.getPayrollConfig().getId();
        salaryCalcContext.isReferenceBenefitData = plan.isReferenceBenefitData();
        salaryCalcContext.payrollPeriodRange = payrollPeriodRange;
        salaryCalcContext.attendancePeriodRange = attendancePeriodRange;

        String companyId = plan.getCompanyId();
        salaryCalcContext.companyId = companyId;
        SalaryPlan mergerTaxPlan = plan.getMergerTaxPlan();
        PayPeriodTerm currentTerm = payPeriod.getCurrentTerm();

        listener.payPeriodId = payPeriod.getId();

        String salaryPlanName = null;

        //This is just a temp logic, Before the calculation, the merge tax salary plan must specify a based on term info .
        if (mergerTaxPlan != null) {

            salaryPlanName = mergerTaxPlan.getName();
            ValidationUtils.rejectIfEmpty(salaryPlanName, null, "方案名称为空");

            Long mergerTaxPlanId = mergerTaxPlan.getId();
            //从历史报表中获取
            HistoryPayPeriod historyPayPeriod = historyPayPeriodRepository.findByCompanyIdAndSalaryPlanSalaryPlanIdAndYearAndMonth(companyId, mergerTaxPlanId, payPeriod.getYear(), payPeriod.getMonth());
            int month = payPeriod.getMonth();
            String monthStr = null;
            if (month < NUM_10) {
                monthStr = "0" + month;
            } else {
                monthStr = "" + month;
            }
            ValidationUtils.rejectIfEmpty(historyPayPeriod, null, String.format("无法在薪资报表中获取%s[%s-%s]的报表", mergerTaxPlan.getName(), payPeriod.getYear(), monthStr));

            List<PayPeriodTerm> terms = historyPayPeriod.getTerms();

            ValidationUtils.rejectIfEmpty(terms, null, String.format("无法在薪资报表中获取%s[%s%s]的报表", mergerTaxPlan.getName(), payPeriod.getYear(), monthStr));
            //目前只获取一个
            PayPeriodTerm currentTerm1 = terms.get(0);
            currentTerm.setBasedOnTerm(currentTerm1);

            salaryCalcContext.basedOnSalaryPlanId = mergerTaxPlanId;
        }
        //

        PayPeriodTerm basedOnTerm = currentTerm.getBasedOnTerm();

        salaryCalcContext.payPeriodTerm =
                new com.ihr360.payperiod.PayPeriodTerm(currentTerm.getId(), currentTerm.getStartOn(), currentTerm.getEndOn(), payPeriod.getYear(), payPeriod.getMonth(), currentTerm.isMergeTaxTerm(), basedOnTerm != null ? basedOnTerm.getId() : null, plan.getName(), salaryPlanName, currentTerm.getMergeSalaryReportId(), currentTerm.getMergeSalaryReportTermSalaryId());
        if (CollectionUtils.isNotEmpty(staffIdList)) {
            salaryCalcContext.staffIdList = staffIdList;
        }
        this.payPeriod = payPeriod;
        return currentTerm;
    }


    public Job salaryCalculationJob(SalaryPlan plan, ActivePayPeriod payPeriod, List<String> staffIdList) {
        threadPoolStatus.showPoolStatus();
        PayPeriodTerm currentTerm = initSalaryContext(plan, payPeriod, staffIdList);

        String taskletName = buildTaskletName(plan.getName() + "[取数并运算]", payPeriod, currentTerm);
        SimpleJobBuilder jobBuilder = jobBuilderFactory
                .get(taskletName)
                .incrementer(new RunIdIncrementer())
                .listener(listener)
                .start(step1())
                .next(step2())
                .next(step3())
                .next(step4())
                .next(step5())
                .next(step6())
                .next(step7())
                .next(step8())
                .next(step9())
                .next(step10());

        if (currentTerm.getBasedOnTerm() != null) {
            jobBuilder.next(step11());
            jobBuilder.next(step12());
        }
        return jobBuilder.next(step13()).build();
    }


    public Job salaryPreCalcJob(SalaryPlan plan, ActivePayPeriod payPeriod, List<String> staffIdList) {
        threadPoolStatus.showPoolStatus();
        PayPeriodTerm currentTerm = initSalaryContext(plan, payPeriod, staffIdList);

        String taskletName = buildTaskletName(plan.getName() + "[取数]", payPeriod, currentTerm);

        return jobBuilderFactory
                .get(taskletName)
                .incrementer(new RunIdIncrementer())
                .listener(listener)
                .start(step1())
                .next(step2())
                .next(step3())
                .next(step4())
                .next(step5())
                .next(step6())
                .next(step7())
                .next(step8())
                .next(step9())
                .next(step13())
                .build();
    }

    private String buildTaskletName(String planName, ActivePayPeriod payPeriod, PayPeriodTerm currentTerm) {
        ValidationUtils.rejectIfEmpty(currentTerm, null, "该账期未设置当前计算批次");
        Calendar instance = Calendar.getInstance();
        instance.setTime(payPeriod.getStartOn());
        return TaskletUtils.getJobName(planName, instance.get(Calendar.YEAR), instance.get(Calendar.MONTH) + 1, currentTerm.getTermIndex());
    }

    public Job salaryCalcWithPreDataJob(SalaryPlan plan, ActivePayPeriod payPeriod, List<String> staffIdList) {
        threadPoolStatus.showPoolStatus();
        PayPeriodTerm currentTerm = initSalaryContext(plan, payPeriod, staffIdList);

        String taskletName = buildTaskletName(plan.getName() + "[核算]", payPeriod, currentTerm);

        SimpleJobBuilder jobBuilder = jobBuilderFactory
                .get(taskletName)
                .incrementer(new RunIdIncrementer())
                .listener(listener)
                .start(step1())
                .next(step2())
                .next(step3())
                .next(step4())
                .next(step5())
                .next(step6())
                .next(step7())
                .next(step8())
                .next(step10());
        if (currentTerm.getBasedOnTerm() != null) {
            jobBuilder.next(step11());
            jobBuilder.next(step12());
        }
        return jobBuilder.next(step13()).build();
    }

    /**
     * 系统薪资项目
     *
     * @return Step
     */
    private Step step1() {
        return stepBuilderFactory.get("获取系统薪资项目定义").allowStartIfComplete(ALLOW_RESTART_IF_COMPLETED)
                .<SystemSalaryField, SystemSalaryField>chunk(CHUNK_SIZE)
                .reader(stepReader.systemSalaryFieldReader(salaryCalcContext.payPeriodId))
                .processor(new SystemSalaryFieldProcessor())
                .listener(loggerListener)
                .writer(salaryCalcContext.systemSalaryFieldsHolder)
                .transactionAttribute(transactionAttribute)
                .build();
    }

    /**
     * 系统薪资项目
     *
     * @return
     */
    private Step step2() {
        return stepBuilderFactory.get("获取引用服务定义").allowStartIfComplete(ALLOW_RESTART_IF_COMPLETED)
                .<List<ReferenceService>, List<ReferenceService>>chunk(CHUNK_SIZE)
                .reader(stepReader.referenceServicesReader())
                .listener(loggerListener)
                .writer(salaryCalcContext.refrenceServicesHolder)
                .transactionAttribute(transactionAttribute)
                .build();
    }

    /**
     * 获取系统变量(税表) 注意生效和失效时间
     * 税表由 地区+缴纳身份+缴纳类型 和生效日期决定
     * 税表的通过提供 <code>纳税身份$税表</code>的引用Code为公式提供引用
     * <p>
     * 为了简便公式配置,税表的地区信息由程序来实现相应过滤
     * <p>
     * 此步骤会准备所有 缴纳身份+缴纳类型的税表数据组织，供员工计算时调用
     *
     * @return
     */
    private Step step3() {
        TaxTableProcessor processor = new TaxTableProcessor(payPeriod.getStartOn());
        return stepBuilderFactory.get("获取税表数据").allowStartIfComplete(ALLOW_RESTART_IF_COMPLETED)
                .<SalaryTaxTableEntity, Pair<String, Map<String, ExprArray>>>chunk(CHUNK_SIZE)
                .reader(stepReader.taxTableReader())
                .processor(processor)
                .listener(loggerListener)
                .writer(salaryCalcContext.taxTableHolder)
                .transactionAttribute(transactionAttribute)
                .build();
    }

    /**
     * 读取工作日历
     *
     * @return
     */
    private Step step4() {
        return stepBuilderFactory.get("读取工作日历").allowStartIfComplete(ALLOW_RESTART_IF_COMPLETED)
                .startLimit(Integer.MAX_VALUE)
                .<List<StatutoryVacation>, WorkingCalendar>chunk(CHUNK_SIZE)
                .reader(stepReader.workingCalendar(payPeriod.unionPeriodRange()))
                .processor(workingCalendarProcessor)
                .listener(loggerListener)
                .writer(salaryCalcContext.workingCalendarHolder)
                .transactionAttribute(transactionAttribute)
                .build();
    }

    /**
     * 读取薪资表
     *
     * @return
     */
    private Step step5() {

        payrollConfigProcessor.setSalaryCalcContext(salaryCalcContext);
        return stepBuilderFactory.get("读取工资表配置信息").allowStartIfComplete(ALLOW_RESTART_IF_COMPLETED)
                .startLimit(Integer.MAX_VALUE)
                .<SimplePayrollConfig, Map<String, SalaryField>>chunk(CHUNK_SIZE)
                .reader(stepReader.payrollConfigReader(salaryCalcContext.companyId, salaryCalcContext.payrollConfigId))
                .processor(payrollConfigProcessor)
                .listener(loggerListener)
                .writer(salaryCalcContext.salaryFieldsHolder)
                .transactionAttribute(transactionAttribute)
                .build();
    }

    /**
     * 设置公司变量(考勤计薪方案:加班规则，病假扣款，事假扣款，入离职调整)进入表达式引擎
     * 一个考勤方案中是可能同时存在两种工时类型
     *
     * @return
     */
    private Step step6() {
        ItemReader<SalaryCalcContext> salaryContextReader = new ItemReader<SalaryCalcContext>() {
            private boolean hasRead = false;

            @Override
            public SalaryCalcContext read() throws Exception {
                if (hasRead) {
                    return null;
                }
                hasRead = true;
                return salaryCalcContext;
            }
        };
        return stepBuilderFactory.get("设置公司考勤计薪方案设置信息进入计算引擎").allowStartIfComplete(ALLOW_RESTART_IF_COMPLETED)
                .<SalaryCalcContext, SalaryDependencyEngine>chunk(CHUNK_SIZE)
                .reader(salaryContextReader)
                .processor(attendancePlanDependencyProcessor)
                .listener(loggerListener)
                .writer(salaryCalcContext.dependencyEngineHolder)
                .transactionAttribute(transactionAttribute)
                .build();
    }

    /**
     * 设置公式进入表达式引擎,同时
     *
     * @return
     */
    private Step step7() {
        ItemReader<SalaryCalcContext> salaryContextReader = new ItemReader<SalaryCalcContext>() {
            private boolean hasRead = false;

            @Override
            public SalaryCalcContext read() throws Exception {
                if (hasRead) {
                    return null;
                }
                hasRead = true;
                return salaryCalcContext;
            }
        };
        return stepBuilderFactory.get("设置公式进入表达式引擎").allowStartIfComplete(ALLOW_RESTART_IF_COMPLETED)
                .<SalaryCalcContext, SalaryDependencyEngine>chunk(CHUNK_SIZE)
                .reader(salaryContextReader)
                .processor(salaryFieldProcessor)
                .listener(loggerListener)
                .writer(salaryCalcContext.dependencyEngineHolder)
                .transactionAttribute(transactionAttribute)
                .build();
    }

    /**
     * 获取方案中的员工id
     *
     * @return
     */
    private Step step8() {
        return stepBuilderFactory.get("获取方案中的人员列表").allowStartIfComplete(ALLOW_RESTART_IF_COMPLETED)
                .<SalaryPlanStaff, String>chunk(CHUNK_SIZE)
                .reader(CollectionUtils.isNotEmpty(salaryCalcContext.staffIdList)
                        ? stepReader.salaryPlanStaffsReader(salaryCalcContext.staffIdList)
                        : stepReader.salaryPlanStaffsReader(salaryCalcContext.salaryPlanId))
                .processor(new SalaryPlanStaffProcessor())
                .listener(loggerListener)
                .writer(salaryCalcContext.salaryPlanStaffListItemWriter)
                .transactionAttribute(transactionAttribute)
                .build();
    }

    /**
     * 分批次获取员工数据（社保公积金，员工基本信息，获取员工薪资体系数据，根据ReferenceApplication中的服务定义获取数据）
     * 前端输入数据哪里来？
     * 对于分段的人，一个人的计算结果要考虑分段信息和汇总信息的保存 SalaryProfileItemsWrapper 目前是满足不了修改的，需要
     * 重新设计
     *
     * @return
     */
    private Step step9() {
        DateRange payrollPeriodRange = DateRange.build(payPeriod.getStartOn(), payPeriod.getEndOn());
        DateRange attendancePeriodRange = DateRange.build(payPeriod.getAttendanceStartOn(), payPeriod.getAttendanceEndOn());

        combineSalaryCalcBasedInfoReader.setSalaryCalcContext(salaryCalcContext);
        combineSalaryCalcBasedInfoReader.setPeriodRanges(payrollPeriodRange, attendancePeriodRange);
        preSalaryCalcProcessor.setSalaryCalcContext(salaryCalcContext);
        preSalaryCalcProcessor.setPeriodRanges(payrollPeriodRange, attendancePeriodRange);
        return stepBuilderFactory.get("获取员工薪资计算数据").allowStartIfComplete(ALLOW_RESTART_IF_COMPLETED)
                .<Map<String, StaffSalaryCalcPeriod>, PreSalaryCalcProcessedResult>chunk(1)
                .reader(combineSalaryCalcBasedInfoReader)
                .processor(preSalaryCalcProcessor)
                .listener(loggerListener)
                .writer(staffSalaryCalcContextWriter)
                .transactionAttribute(transactionAttribute)
                .build();
    }

    /**
     * 根据取数获取的数据，计算员工薪资
     *
     * @return
     */
    private Step step10() {

        salaryCalcContextInfoReader.setSalaryCalcContext(salaryCalcContext);
        salaryCalcProcessor.setSalaryCalcContext(salaryCalcContext);

        return stepBuilderFactory.get("员工计算").allowStartIfComplete(ALLOW_RESTART_IF_COMPLETED)
                .<List<StaffPayPeriodTermCalcContext>, Map<String, StaffSalaryCalcPeriod>>chunk(1)
                .reader(salaryCalcContextInfoReader)
                .processor(salaryCalcProcessor)
                .listener(loggerListener)
                .writer(calcedStaffSalaryInfoWriter)
                .transactionAttribute(transactionAttribute)
                .build();
    }


    /**
     * 生成合并计税报表（仅供合并计税条件下使用）
     *
     * @return
     */
    private Step step11() {
        return stepBuilderFactory.get("生成合并计税汇总定义").allowStartIfComplete(ALLOW_RESTART_IF_COMPLETED)
                .<SalaryCalcContext, SalaryCalcContext>chunk(1)
                .reader(stepReader.salaryCalcContextItemReader(salaryCalcContext))
                .processor(mergeTaxSalaryReportMetaProcessor)
                .listener(loggerListener)
                .transactionAttribute(transactionAttribute)
                .build();
    }

    /**
     * 生成合并计税报表（仅供合并计税条件下使用）
     *
     * @return
     */
    private Step step12() {
        mergeTaxSalaryReportDataProcessor.setSalaryCalcContext(salaryCalcContext);
        return stepBuilderFactory.get("生成合并计税汇总报表").allowStartIfComplete(ALLOW_RESTART_IF_COMPLETED)
                .<List<String>, List<MergeReportStaffSalary>>chunk(1)
                .reader(CollectionUtils.isNotEmpty(salaryCalcContext.staffIdList)
                        ? stepReader.mergedSalaryPlanStaffIdsReader(salaryCalcContext.staffIdList)
                        : stepReader.mergedSalaryPlanStaffsReader(ImmutableList.of(salaryCalcContext.salaryPlanId, salaryCalcContext.basedOnSalaryPlanId)))
                .processor(mergeTaxSalaryReportDataProcessor)
                .listener(loggerListener)
                .writer(mergeTaxSalaryReportDataWriter)
                .transactionAttribute(transactionAttribute)
                .build();
    }

    /**
     * 回写状态到active payperiod
     *
     * @return
     */
    private Step step13() {
        return stepBuilderFactory.get("更新账期状态").allowStartIfComplete(ALLOW_RESTART_IF_COMPLETED)
                .<ActivePayPeriod, ActivePayPeriod>chunk(CHUNK_SIZE)
                .reader(stepReader.activePayPeriodReader(salaryCalcContext.payPeriodId))
                .listener(loggerListener)
                .writer(activePayPeriodWriter)
                .build();
    }


    private DefaultTransactionAttribute transactionAttribute = new DefaultTransactionAttribute() {
        @Override
        public boolean rollbackOn(Throwable ex) {
            if (ex instanceof Ihr360Exception) {
                return false;
            }
            return (ex instanceof RuntimeException || ex instanceof Error);
        }
    };

}